package FutureAndPromise.Demo03_NettyFuture;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import lombok.extern.slf4j.Slf4j;

import java.nio.charset.StandardCharsets;

/**
 * Demo03 - Netty ChannelFuture 详解 - 客户端
 * 
 * 学习目标：
 * 1. 理解 Netty Future 和 JDK Future 的区别
 * 2. 掌握 ChannelFuture 的状态查询方法
 * 3. 学习使用监听器处理异步结果
 * 4. 了解 sync() 和 await() 的使用场景
 */
@Slf4j
public class Client {

    public static void main(String[] args) throws Exception {
        log.info("=== Netty ChannelFuture 详解 - 客户端 ===\n");
        log.info("请先启动 Server.java\n");
        
        demo1_FutureStates();
        demo2_ListenerBasics();
        demo3_MultipleListeners();
        demo4_PreDefinedListeners();
        demo5_SyncVsAwait();
        demo6_BestPractices();
        
        Thread.sleep(1000);
        System.exit(0);
    }

    /**
     * Demo 1: ChannelFuture 的状态
     */
    private static void demo1_FutureStates() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 1: ChannelFuture 的状态");
        log.info("└─────────────────────────────────────────\n");
        
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new SimpleHandler());
                    }
                });
            
            // 1. 连接操作返回 ChannelFuture
            log.info("1. 发起连接...");
            ChannelFuture connectFuture = b.connect("localhost", 9000);
            
            // 2. 此时连接可能还没建立
            log.info("2. 立即返回 ChannelFuture");
            log.info("   isDone: {}", connectFuture.isDone());
            log.info("   isSuccess: {}", connectFuture.isSuccess());
            log.info("   isCancelled: {}", connectFuture.isCancelled());
            
            // 3. 等待连接完成
            connectFuture.sync();
            log.info("3. 连接完成后");
            log.info("   isDone: {}", connectFuture.isDone());
            log.info("   isSuccess: {}", connectFuture.isSuccess());
            log.info("   isCancellable: {}", connectFuture.isCancellable());
            
            Channel channel = connectFuture.channel();
            
            // 4. 写操作也返回 ChannelFuture
            log.info("\n4. 发送数据...");
            ChannelFuture writeFuture = channel.writeAndFlush(
                Unpooled.copiedBuffer("Test message\n", StandardCharsets.UTF_8)
            );
            
            log.info("   写操作立即返回");
            log.info("   isDone: {}", writeFuture.isDone());
            
            writeFuture.sync();
            log.info("   写操作完成");
            log.info("   isDone: {}", writeFuture.isDone());
            log.info("   isSuccess: {}", writeFuture.isSuccess());
            
            Thread.sleep(500);
            channel.close().sync();
            
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * Demo 2: 监听器基础
     */
    private static void demo2_ListenerBasics() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 2: 监听器基础");
        log.info("└─────────────────────────────────────────\n");
        
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new SimpleHandler());
                    }
                });
            
            // 方式 1: 匿名内部类
            log.info("1. 使用匿名内部类添加监听器");
            ChannelFuture connectFuture = b.connect("localhost", 9000);
            connectFuture.addListener(new ChannelFutureListener() {
                @Override
                public void operationComplete(ChannelFuture future) {
                    if (future.isSuccess()) {
                        log.info("   ✅ [监听器] 连接成功！");
                        Channel channel = future.channel();
                        log.info("   [监听器] 本地地址: {}", channel.localAddress());
                        log.info("   [监听器] 远程地址: {}", channel.remoteAddress());
                    } else {
                        log.error("   ❌ [监听器] 连接失败: {}", future.cause().getMessage());
                    }
                }
            });
            
            connectFuture.sync();
            Thread.sleep(500);
            
            // 方式 2: Lambda 表达式
            log.info("\n2. 使用 Lambda 表达式");
            Channel channel = connectFuture.channel();
            
            ChannelFuture writeFuture = channel.writeAndFlush(
                Unpooled.copiedBuffer("Lambda test\n", StandardCharsets.UTF_8)
            );
            
            writeFuture.addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.info("   ✅ [Lambda] 数据发送成功");
                } else {
                    log.error("   ❌ [Lambda] 数据发送失败");
                }
            });
            
            Thread.sleep(500);
            channel.close().sync();
            
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * Demo 3: 多个监听器
     */
    private static void demo3_MultipleListeners() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 3: 链式添加多个监听器");
        log.info("└─────────────────────────────────────────\n");
        
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new SimpleHandler());
                    }
                });
            
            log.info("为同一个 Future 添加多个监听器（按添加顺序执行）");
            
            ChannelFuture connectFuture = b.connect("localhost", 9000);
            
            // 链式添加多个监听器
            connectFuture
                .addListener((ChannelFutureListener) future -> {
                    log.info("   [监听器 1] 记录连接日志");
                })
                .addListener((ChannelFutureListener) future -> {
                    log.info("   [监听器 2] 发送统计信息");
                })
                .addListener((ChannelFutureListener) future -> {
                    log.info("   [监听器 3] 初始化会话");
                    if (future.isSuccess()) {
                        Channel channel = future.channel();
                        channel.writeAndFlush(
                            Unpooled.copiedBuffer("Init message\n", StandardCharsets.UTF_8)
                        );
                    }
                })
                .addListener((ChannelFutureListener) future -> {
                    log.info("   [监听器 4] 所有初始化完成");
                });
            
            connectFuture.sync();
            Thread.sleep(500);
            connectFuture.channel().close().sync();
            
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * Demo 4: 预定义监听器
     */
    private static void demo4_PreDefinedListeners() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 4: 预定义监听器");
        log.info("└─────────────────────────────────────────\n");
        
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new SimpleHandler());
                    }
                });
            
            // 1. CLOSE - 操作完成后关闭连接
            log.info("1. ChannelFutureListener.CLOSE - 完成后关闭");
            ChannelFuture connectFuture1 = b.connect("localhost", 9000).sync();
            Channel channel1 = connectFuture1.channel();
            
            channel1.writeAndFlush(
                Unpooled.copiedBuffer("Final message\n", StandardCharsets.UTF_8)
            ).addListener(ChannelFutureListener.CLOSE);
            
            log.info("   消息发送完成后会自动关闭连接");
            Thread.sleep(500);
            
            // 2. CLOSE_ON_FAILURE - 失败时关闭
            log.info("\n2. ChannelFutureListener.CLOSE_ON_FAILURE - 失败时关闭");
            ChannelFuture connectFuture2 = b.connect("localhost", 9000).sync();
            Channel channel2 = connectFuture2.channel();
            
            channel2.writeAndFlush(
                Unpooled.copiedBuffer("Test\n", StandardCharsets.UTF_8)
            ).addListener(ChannelFutureListener.CLOSE_ON_FAILURE)
             .addListener(future -> {
                 if (future.isSuccess()) {
                     log.info("   ✅ 发送成功，连接保持打开");
                 }
             });
            
            Thread.sleep(500);
            channel2.close().sync();
            
            // 3. FIRE_EXCEPTION_ON_FAILURE - 失败时触发异常
            log.info("\n3. ChannelFutureListener.FIRE_EXCEPTION_ON_FAILURE");
            log.info("   失败时会在 Pipeline 中触发 exceptionCaught 事件");
            
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * Demo 5: sync() vs await()
     */
    private static void demo5_SyncVsAwait() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 5: sync() vs await()");
        log.info("└─────────────────────────────────────────\n");
        
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new SimpleHandler());
                    }
                });
            
            // 1. sync() - 失败时抛出异常
            log.info("1. sync() - 连接失败的端口");
            try {
                ChannelFuture future = b.connect("localhost", 9999);
                future.sync();  // 会抛出异常
            } catch (Exception e) {
                log.error("   ❌ sync() 抛出异常: {}", e.getCause().getMessage());
            }
            
            // 2. await() - 失败不抛异常
            log.info("\n2. await() - 连接失败的端口");
            ChannelFuture future = b.connect("localhost", 9999);
            future.await();  // 不会抛异常
            
            if (future.isSuccess()) {
                log.info("   ✅ 连接成功");
            } else {
                log.error("   ❌ await() 不抛异常，需手动检查: {}", 
                    future.cause().getMessage());
            }
            
            // 3. awaitUninterruptibly() - 不响应中断
            log.info("\n3. awaitUninterruptibly() - 不响应线程中断");
            ChannelFuture future2 = b.connect("localhost", 9000);
            future2.awaitUninterruptibly();
            log.info("   等待完成（即使线程被中断也会继续等待）");
            
            if (future2.isSuccess()) {
                future2.channel().close().sync();
            }
            
        } finally {
            group.shutdownGracefully();
        }
    }

    /**
     * Demo 6: 最佳实践
     */
    private static void demo6_BestPractices() throws Exception {
        log.info("\n┌─────────────────────────────────────────");
        log.info("│ Demo 6: 最佳实践");
        log.info("└─────────────────────────────────────────\n");
        
        EventLoopGroup group = new NioEventLoopGroup();
        try {
            Bootstrap b = new Bootstrap();
            b.group(group)
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) {
                        ch.pipeline().addLast(new SimpleHandler());
                    }
                });
            
            // 实践 1: 连接成功后再发送数据
            log.info("实践 1: 确保连接成功后再操作");
            b.connect("localhost", 9000).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    log.info("   ✅ 连接成功，开始发送数据");
                    Channel channel = future.channel();
                    
                    channel.writeAndFlush(
                        Unpooled.copiedBuffer("Practice 1\n", StandardCharsets.UTF_8)
                    ).addListener(writeFuture -> {
                        if (writeFuture.isSuccess()) {
                            log.info("   ✅ 数据发送成功");
                        } else {
                            log.error("   ❌ 数据发送失败", writeFuture.cause());
                        }
                    });
                } else {
                    log.error("   ❌ 连接失败: {}", future.cause().getMessage());
                }
            });
            
            Thread.sleep(500);
            
            // 实践 2: 发送完成后关闭连接
            log.info("\n实践 2: 发送完成后优雅关闭");
            ChannelFuture connectFuture = b.connect("localhost", 9000).sync();
            Channel channel = connectFuture.channel();
            
            channel.writeAndFlush(
                Unpooled.copiedBuffer("Practice 2\n", StandardCharsets.UTF_8)
            ).addListener(ChannelFutureListener.CLOSE);
            
            log.info("   消息发送完成后会自动关闭连接");
            
            Thread.sleep(500);
            
            // 实践 3: 完善的异常处理
            log.info("\n实践 3: 完善的异常处理");
            b.connect("localhost", 9000).addListener((ChannelFutureListener) future -> {
                if (future.isSuccess()) {
                    Channel ch = future.channel();
                    log.info("   ✅ 连接成功: {}", ch.remoteAddress());
                    
                    ch.writeAndFlush(
                        Unpooled.copiedBuffer("Practice 3\n", StandardCharsets.UTF_8)
                    ).addListener(writeFuture -> {
                        if (writeFuture.isSuccess()) {
                            log.info("   ✅ 发送成功");
                            ch.close();
                        } else {
                            log.error("   ❌ 发送失败，关闭连接", writeFuture.cause());
                            ch.close();
                        }
                    });
                } else {
                    log.error("   ❌ 连接失败: {}", future.cause().getMessage());
                    // 这里可以进行重连等操作
                }
            });
            
            Thread.sleep(1000);
            
        } finally {
            group.shutdownGracefully();
        }
    }

    // 简单的处理器
    @Slf4j
    static class SimpleHandler extends ChannelInboundHandlerAdapter {
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) {
            ByteBuf buf = (ByteBuf) msg;
            log.info("[客户端] 收到回复: {}", buf.toString(StandardCharsets.UTF_8).trim());
            buf.release();
        }

        @Override
        public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
            log.error("[客户端] 异常", cause);
            ctx.close();
        }
    }
}

/*
=== 运行结果分析 ===

1. Netty Future vs JDK Future：

   Netty Future 扩展了 JDK Future：
   ✅ 继承 java.util.concurrent.Future
   ✅ 添加了 Listener 机制
   ✅ 提供了更多状态查询方法
   ✅ sync() 和 await() 方法
   ✅ 可以添加多个监听器

2. ChannelFuture 的状态：

   isDone()：操作是否完成
   isSuccess()：操作是否成功
   isCancelled()：操作是否被取消
   isCancellable()：操作是否可以取消
   cause()：失败原因

3. 使用监听器的优势：

   ✅ 不阻塞线程
   ✅ 在 EventLoop 线程中执行
   ✅ 可以链式调用
   ✅ 异常处理方便

4. sync() vs await()：

   sync()：
   - 阻塞等待
   - 失败时抛出异常
   - 常用于启动时

   await()：
   - 阻塞等待
   - 失败不抛异常
   - 需要手动检查结果

   awaitUninterruptibly()：
   - 阻塞等待
   - 不响应线程中断

=== 关键知识点 ===

1. Netty 所有 I/O 操作都返回 ChannelFuture
2. 推荐使用监听器，避免阻塞
3. 可以为同一个 Future 添加多个监听器
4. 预定义监听器简化常见操作
5. 不要在 EventLoop 线程中使用 sync()

=== 最佳实践 ===

✅ 使用监听器处理异步结果
✅ 连接成功后再发送数据
✅ 完善的异常处理
✅ 使用 CLOSE 监听器优雅关闭
✅ 避免在 EventLoop 线程中 sync()

❌ 不要忽略 isSuccess() 检查
❌ 不要在 EventLoop 中阻塞
❌ 不要忘记处理失败情况
*/

